package com.lkx.rocketmq.consumer;


import com.alibaba.fastjson.JSON;
import com.lkx.dao.OrderMapper;
import com.lkx.domain.DemoGoodsStatistics;
import com.lkx.domain.GoodsStatistics;
import com.lkx.domain.Order;
import com.lkx.service.GoodsStatisticsService;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.List;

/**
 * ClassName: MQConsumeMsgListenerProcessor
 * Function:  TODO
 * Date:      2019-06-19 13:30
 * author     likaixuan
 * version    V1.0
 */
@Component
public class MQConsumeMsgListenerProcessor  implements MessageListenerConcurrently {
    private static final Logger logger = LoggerFactory.getLogger(MQConsumeMsgListenerProcessor.class);

    @Resource
    private OrderMapper orderMapper;
    @Autowired
    private GoodsStatisticsService goodsStatisticsService;

    /**
     *  默认msgs里只有一条消息，可以通过设置consumeMessageBatchMaxSize参数来批量接收消息<br/>
     *  不要抛异常，如果没有return CONSUME_SUCCESS ，consumer会重新消费该消息，直到return CONSUME_SUCCESS
     */
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        if(CollectionUtils.isEmpty(msgs)){
            logger.info("接受到的消息为空，不处理，直接返回成功");
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        MessageExt messageExt = msgs.get(0);
        logger.info("接受到的消息为："+messageExt.toString());
        if(messageExt.getTopic().equals("DemoTopic")){
            if(messageExt.getTags().equals("DemoTag")){
                //TODO 判断该消息是否重复消费（RocketMQ不保证消息不重复，如果你的业务需要保证严格的不重复消息，需要你自己在业务端去重）
                //TODO 获取该消息重试次数
                int reconsume = messageExt.getReconsumeTimes();
                String str= new String (messageExt.getBody());
                Order order = JSON.parseObject(str,Order.class);
                orderMapper.insert(order);
                //System.out.println(reconsume+":"+messageExt.getTopic()+":"+str);
                if(reconsume ==3){//消息已经重试了3次，如果不需要再次消费，则返回成功
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                //TODO 处理对应的业务逻辑
            }else if(messageExt.getTags().equals("DemoTag1")){
                int reconsume = messageExt.getReconsumeTimes();
                String str= new String (messageExt.getBody());
                DemoGoodsStatistics goodsStatistics = JSON.parseObject(str,DemoGoodsStatistics.class);
                logger.info(reconsume+":"+messageExt.getTopic()+":"+str);
                if(reconsume ==3){//消息已经重试了3次，如果不需要再次消费，则返回成功
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                try {
                    goodsStatisticsService.testMysqlLock(goodsStatistics.getGoodsId1(),goodsStatistics.getGoodsId2());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
        // 如果没有return success ，consumer会重新消费该消息，直到return success
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
